package com.langangkj.tools.redis.handler;

import com.langangkj.tools.redis.handler.meta.RedisKeyMeta;
import com.langangkj.tools.redis.handler.provider.RedisSyncKeyProvider;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.util.Optional;
import java.util.concurrent.Executor;

/**
 * @author ZhouChuGang
 * @version 1.0
 * @project redis-sync-tools
 * @date 2020/4/26 9:34
 * @Description redis同步处理类
 */
@Slf4j
@Component
public class RedisSyncHandler implements CommandLineRunner, DisposableBean {

    public RedisSyncHandler() {
        log.info("处理者启动。。。");
    }

    @Autowired
    private RedisSyncKeyProvider redisSyncKeyProvider;

    @Autowired
    private Executor executor;

    /**
     * 运行标记
     */
    private volatile boolean runFlag = true;

    @Override
    public void run(String... args) throws Exception {
        for (int i = 0; i < 10; i++) {
            executor.execute(getRunnable());
        }
    }

    /**
     * 得到运行对象
     *
     * @return
     */
    private Runnable getRunnable() {
        return () -> {
            while (runFlag) {
                try {
                    RedisKeyMeta take = redisSyncKeyProvider.getBlockingQueue().take();
                    Optional.ofNullable(take).ifPresent(redisKeyMeta -> {
                        Optional.ofNullable(redisSyncKeyProvider.getTypeRedisKeyMetaHandlerMap().get(redisKeyMeta.getDataType()))
                                .ifPresent(handler -> {
                                    log.info("处理 key mate {} ", redisKeyMeta);
                                    handler.handler(redisKeyMeta);
                                });
                    });
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
    }

    @Override
    public void destroy() throws Exception {
        log.info("关闭线程池");
        redisSyncKeyProvider.getBlockingQueue().clear();
        runFlag = false;
    }
}
